package com.zhang.storm.pv;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;

public class PvMain {
    public static void main(String[] args) {

        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("PvSpout", new PvSpout(), 1);
        builder.setBolt("PvBolt", new PvBolt(), 4).shuffleGrouping("PvSpout");
        builder.setBolt("PvSumBolt", new PvSumBolt(), 1).shuffleGrouping("PvBolt");

        Config config = new Config();
        config.setNumWorkers(2);

        if (args.length > 0) {
            try {
                StormSubmitter.submitTopology("PV", config, builder.createTopology());
            } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
                e.printStackTrace();
            }
        } else {
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology("PV", config, builder.createTopology());
        }

    }
}
